package com.atguigu.gmall.realtime.app

import java.text.SimpleDateFormat
import java.time.{LocalDate, Period}
import java.util
import java.util.Date

import com.alibaba.fastjson.serializer.SerializeConfig
import com.alibaba.fastjson.{JSON, JSONObject}
import com.atguigu.gmall.realtime.bean.{OrderDetail, OrderInfo, OrderWide, PageLog}
import com.atguigu.gmall.realtime.util.{MyEsUtil, MyKafkaUtil, OffsetManager, RedisUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import scala.collection.mutable.ListBuffer


// 下单业务宽表
object DwOrderApp {


  //0 环境 ssc
  //1  读取偏移量  * 2
  //  2 接收kafka数据  *2
  //3  提取偏移量结束点  *2
  //4  转换结构 msg->   PageLog  *2
  // 5 合并维度
  // 6 双流 join
  //  7写入es
  //8  提交偏移量
  def main(args: Array[String]): Unit = {
    //0 环境 ssc
    val sparkConf: SparkConf = new SparkConf().setAppName("dw_order_app").setMaster("local[4]")
    val ssc = new StreamingContext(sparkConf,Seconds(5))

    val orderInfoTopic="DWD_ORDER_INFO_I"
    val orderDetailTopic="DWD_ORDER_DETAIL_I"
    val  groupId="dw_order_group"
    //1  读取偏移量  * 2
    val orderInfoOffsetMap: Map[TopicPartition, Long] = OffsetManager.getOffset(orderInfoTopic,groupId)
    val orderDetailOffsetMap: Map[TopicPartition, Long] = OffsetManager.getOffset(orderDetailTopic,groupId)

    //  2 接收kafka数据
    //order_info
    var orderInfoInputDstream: InputDStream[ConsumerRecord[String, String]]=null
    if(orderInfoOffsetMap!=null && orderInfoOffsetMap.size>0){
      orderInfoInputDstream  = MyKafkaUtil.getKafkaStream(orderInfoTopic,ssc,orderInfoOffsetMap ,groupId)
    }else{
      orderInfoInputDstream  = MyKafkaUtil.getKafkaStream(orderInfoTopic,ssc ,groupId)
    }
    //order_detail
    var orderDetailInputDstream: InputDStream[ConsumerRecord[String, String]]=null
    if(orderDetailOffsetMap!=null && orderDetailOffsetMap.size>0){
      orderDetailInputDstream  = MyKafkaUtil.getKafkaStream(orderDetailTopic,ssc,orderDetailOffsetMap ,groupId)
    }else{
      orderDetailInputDstream  = MyKafkaUtil.getKafkaStream(orderDetailTopic,ssc ,groupId)
    }

    //3  提取偏移量结束点
    //order_info
    var orderInfoOffsetRanges: Array[OffsetRange]=null  //driver
    val orderInfoInputOffsetDstream: DStream[ConsumerRecord[String, String]] = orderInfoInputDstream.transform { rdd =>
      orderInfoOffsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges   // 每批次执行一次
      rdd
    }
    //order_detail
    var orderDetailOffsetRanges: Array[OffsetRange]=null  //driver
    val orderDetailInputOffsetDstream: DStream[ConsumerRecord[String, String]] = orderDetailInputDstream.transform { rdd =>
      orderDetailOffsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges   // 每批次执行一次
      rdd
    }
    //4  转换结构 msg->  OrderInfo  OrderDetail
    val orderInfoDstream: DStream[OrderInfo] = orderInfoInputOffsetDstream.map { record =>
      val jsonString: String = record.value()
      val orderInfo: OrderInfo = JSON.parseObject(jsonString, classOf[OrderInfo])
      orderInfo
    }



    val orderDetailDstream: DStream[OrderDetail] = orderDetailInputOffsetDstream.map { record =>
      val jsonString: String = record.value()
      val orderDetail: OrderDetail = JSON.parseObject(jsonString, classOf[OrderDetail])
      orderDetail
    }
    //5 维度合并 (作业)

    // orderInfoDstream===>  orderInfoWithDimDstream[OrderInfo]  //补充用户年龄性别 地区
    val orderInfoWithDimDstream:DStream[OrderInfo]  = orderInfoDstream.mapPartitions{orderInfoItr=>
      val jedisClient: Jedis = RedisUtil.getJedisClient
      val orderInfoList: List[OrderInfo] = orderInfoItr.toList
      for (orderInfo <- orderInfoList) {
        //补充维度信息 ： 地区
        val provinceKey = "DIM:BASE_PROVINCE:" + orderInfo.province_id
        val provinceJson: String = jedisClient.get(provinceKey)
        val provinceJsonObj: JSONObject = JSON.parseObject(provinceJson)
        orderInfo.province_name = provinceJsonObj.getString("name")
        orderInfo.province_area_code = provinceJsonObj.getString("area_code") // ali datav  quickbi baidu suger
        orderInfo.province_3166_2_code = provinceJsonObj.getString("iso_3166_2") // kibana
        orderInfo.province_iso_code = provinceJsonObj.getString("iso_code") // superset

        // 补充维度信息: 用户基本信息  :年龄 性别
        val userInfoKey = "DIM:USER_INFO:" + orderInfo.user_id
        val userInfoJson: String = jedisClient.get(userInfoKey)
        val userInfoJsonObj: JSONObject = JSON.parseObject(userInfoJson)
        orderInfo.user_gender = userInfoJsonObj.getString("gender")
        val birthday = userInfoJsonObj.getString("birthday")
        val birthdayDate2: LocalDate = LocalDate.parse(birthday)
        val now: LocalDate = LocalDate.now()
        var period = Period.between(birthdayDate2, now);
        val age: Int = period.getYears
        orderInfo.user_age = age.toInt
        //补充日期字段 create_date  和 create_hour
        val dateTimeArr: Array[String] = orderInfo.create_time.split(" ")
        orderInfo.create_date= dateTimeArr(0)
        orderInfo.create_hour= dateTimeArr(1).split(":")(0)

      }
      jedisClient.close()
      orderInfoList.toIterator
    }
    // orderDetailDstream==>orderDetailWithDimDstream[OrderDetail] //补充品牌 、品类

    //6 双流join
    // 1 必须是 Dstream[K,V] 和 Dstream[K,V] 才能join
    val orderInfoWithKeyDstream: DStream[(Long, OrderInfo)] = orderInfoWithDimDstream.map(orderInfo=>(orderInfo.id,orderInfo))
    val orderDetailWithKeyDstream: DStream[(Long, OrderDetail)] = orderDetailDstream.map(orderDetail=>(orderDetail.order_id,orderDetail))
    // val orderJoinDstream: DStream[(Long, (OrderInfo, OrderDetail))] = orderInfoWithKeyDstream.join(orderDetailWithKeyDstream)

    val orderJoinDstream: DStream[(Long, (Option[OrderInfo], Option[OrderDetail]))] = orderInfoWithKeyDstream.fullOuterJoin(orderDetailWithKeyDstream)

    val orderWideDStream: DStream[OrderWide] = orderJoinDstream.flatMap { case (orderId, (orderInfoOpt, orderDetailOpt)) =>
      //判断 主表在否
      val orderWideList: ListBuffer[OrderWide] = ListBuffer[OrderWide]()
      val jedisClient: Jedis = RedisUtil.getJedisClient
      if (orderInfoOpt != None) {
        val orderInfo: OrderInfo = orderInfoOpt.get
        //1  主表在
        //1.1  判断是否从表在
        if (orderDetailOpt != None) {
          val orderDetail: OrderDetail = orderDetailOpt.get
          //如果在把两个对象合并   // new OrderWide(orderInfo ,orderDetail)
          val orderWide = new OrderWide(orderInfo, orderDetail)
          orderWideList.append(orderWide)
        }
        //1.2  主表写缓存
        //type ？ string   1个key对应一个对象  key ? ORDER_JOIN:ORDER_INFO:[ID]   value? orderInfoJson   field/score? 无
        // 写入api？set   读取api ?  get  过期时间？ 小时-天 都可以  24小时
        val orderInfoKey = s"ORDER_JOIN:ORDER_INFO:${orderInfo.id}"
        val orderInfoJson = JSON.toJSONString(orderInfo, new SerializeConfig(true))
        jedisClient.setex(orderInfoKey, 3600 * 24, orderInfoJson)

        //1.3 主表读缓存
        //如果命中把两个对象合并
        //从表缓存
        //type ？  set     key ? ORDER_JOIN:ORDER_DETAIL:[ORDERID]      value? orderDetailJson 多个...    field/score? 无
        // 写入api？  sadd   读取api ? smembers    过期时间？ 小时-天 都可以  24小时
        val orderDetailKey = s"ORDER_JOIN:ORDER_DETAIL:${orderInfo.id}"
        val orderDetailJsonSet: util.Set[String] = jedisClient.smembers(orderDetailKey)
        import scala.collection.JavaConverters._
        if(orderDetailJsonSet!=null&&orderDetailJsonSet.size()>0){
          for (orderDetailJson <- orderDetailJsonSet.asScala) {
            val orderDetail: OrderDetail = JSON.parseObject(orderDetailJson, classOf[OrderDetail])
            val orderWide = new OrderWide(orderInfo, orderDetail)
            orderWideList.append(orderWide)
          }
        }
      } else {
        //2  主表不在  从表在
        val orderDetail: OrderDetail = orderDetailOpt.get
        //2.1 从表读缓存
        //如果命中把两个对象合并
        val orderInfoKey = s"ORDER_JOIN:ORDER_INFO:${orderDetail.order_id}"
        val orderInfoJson: String = jedisClient.get(orderInfoKey)
        if (orderInfoJson != null && orderInfoJson.length > 0) {
          val orderInfo: OrderInfo = JSON.parseObject(orderInfoJson, classOf[OrderInfo])
          val orderWide = new OrderWide(orderInfo, orderDetail)
          orderWideList.append(orderWide)
        }
        //2.2  从表写缓存
        val orderDetailKey = s"ORDER_JOIN:ORDER_DETAIL:${orderDetail.order_id}"
        val orderDetailJson = JSON.toJSONString(orderDetail, new SerializeConfig(true))
        jedisClient.sadd(orderDetailKey, orderDetailJson)
        jedisClient.expire(orderDetailKey, 3600 * 24)

      }

      jedisClient.close()
      orderWideList
    }
    orderWideDStream.cache()// 流中如果有两个行动算子  行动算子前要加cache
    orderWideDStream.print(10000)


    orderWideDStream.foreachRDD{rdd=>
      rdd.foreachPartition{ orderWideItr=>
        //1executor
        val orderWideList: List[OrderWide] = orderWideItr.toList
        if(orderWideList.size>0){
          val orderWideFirst: OrderWide = orderWideList(0)
          val dt: String = orderWideFirst.create_date
          val orderWideWithIdList: List[(OrderWide, String)] = orderWideList.map(orderWide=>(orderWide,orderWide.detail_id.toString))

          val indexName=s"gmall0701_order_wide_$dt"
          MyEsUtil.bulkSave(orderWideWithIdList,indexName)
        }


      }
      //2  driver
      //提交偏移量
      OffsetManager.saveOffset(orderInfoTopic,groupId,orderInfoOffsetRanges)
      OffsetManager.saveOffset(orderDetailTopic,groupId,orderDetailOffsetRanges)

    }



    // orderJoinDstream.print(10000)

    ssc.start()
    ssc.awaitTermination()

  }

}
